package com.bw.yk01;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;


//OrderWide
//OrderWide
public class AsyncIOFunc extends RichAsyncFunction<OrderWide,OrderWide> {

    Jedis jedis = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("hadoop-single");
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }

    @Override
    public void asyncInvoke(OrderWide input, ResultFuture<OrderWide> resultFuture) throws Exception {
        //从订单明细表获取sku_id,再从redis获取信息
        String sku_id = input.getOrderDetail().getSku_id();
        String skujo = jedis.get("sku_info:" + sku_id);
        SkuInfo skuInfo = JSON.parseObject(skujo, SkuInfo.class);
        input.setSkuInfo(skuInfo);
        //从订单明表获取user_id,再从redis获取信息
        String user_id = input.getOrderInfo().getUser_id();
        String userjo = jedis.get("user_info:" + user_id);
        UserInfo userInfo = JSON.parseObject(userjo, UserInfo.class);
        input.setUserInfo(userInfo);
        //从订单明表获取province_id,再从redis获取信息
        String province_id = input.getOrderInfo().getProvince_id();
        String provincejo = jedis.get("base_province:" + province_id);
        BaseProvince baseProvince = JSON.parseObject(provincejo, BaseProvince.class);
        input.setBaseProvince(baseProvince);
        //收集返回
        resultFuture.complete(Collections.singleton(input));
    }
}
